package org.ossean.gather.process;
/**
 * @author 李乾坤 2016年12月5号第一次修改。修改内容如下：
 * 1.将获取tag部分抽离为一个单独的可以接收参数的函数handleTags
 * 2.改变了数据汇总策略。放弃之前把记录全放到set中的方法。采用如下策略：
 * 1>查询pk_control_*表，查看当前帖子或者项目是否已经有固定Id
 * 2>查询到固定Id值（sameId），则依据sameId更新数据库中的内容。未查到固定Id进行3>
 * 3>将当前项目或帖子依据url_md5插入pk_control_*表以生成固定Id（newId），然后根据newId执行插入操作。
 * */
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import javax.annotation.Resource;

import org.apache.log4j.Logger;
import org.ossean.gather.model.Configure;
import org.ossean.gather.model.GatherProject;
import org.ossean.gather.model.JobRequirement;
import org.ossean.gather.model.PKControlPosts;
import org.ossean.gather.model.PKControlProjects;
import org.ossean.gather.model.RelativeMemo;
import org.ossean.gather.model.Taggings;
import org.ossean.gather.sourceDao.GatherDao;
import org.ossean.gather.sourceDao.PKControlPostsDao;
import org.ossean.gather.sourceDao.PKControlProjectsDao;
import org.ossean.gather.targetDao.PointerDao;
import org.ossean.gather.targetDao.TargetDao;
import org.springframework.context.annotation.Scope;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

@Component("gatherThreadNew")
@Scope("prototype")
public class GatherThreadNew implements Runnable {
	private static Logger logger = Logger.getLogger(GatherThread.class);
	private Configure conf;

	//注解的方式生成bean并获取汇总操作对象
	@Resource
	private GatherDao gatherDao;
	@Resource
	private PointerDao pointerDao;
	//控制帖子的操作
	@Resource
	private PKControlPostsDao pkControlPostsDao;
	//汇总表操作对象
	@Resource
	private TargetDao targetDao;
	//项目控制表操作
	@Resource
	private PKControlProjectsDao pkControlProjectsDao;

	private int idsBegin; // 转移开始Id值
	private int idsEnd; // 转移结束Id值
	private int idsIncrement;// 每次转移的Id量

	private int beginId;
	private int endId;

	private String sourceTableName;
	private String pkControlPostsTableName = "pk_control_posts";
	private String pkControlProjectsTableName = "pk_control_projects";

	private String gatherPostsTableName = "relative_memos";
	private int maxId;
	
	private String memoTaggingsTableName = "memo_taggings";
	private String tagsTableName = "tags";

	public void setParameters(Configure conf, String sourceTableName) {
		this.conf = conf;
		this.sourceTableName = sourceTableName;
	}

	// 读指针
	public int readPointer(String table, String source, String target) {
		int pointer = 1;
		try {
			//调用pointerDao对象读取数据库的pointer表获取上次汇总的位置
			pointer = pointerDao.getPointer(table, source, target);
		} catch (Exception e) {
			// 表示表中没有数据
			logger.info("No such pointer! Create one");
			//表不存在的话就调用pointerDao中的函数创建pointer的表并插入1
			pointerDao.insertPointer(table, source, target, 1);
		}
		return pointer;
	}

	@Override
	public void run(){
		//获取系统当年时间
		//long start = System.currentTimeMillis();
		
		//设置当前进程数据源的表名
		Thread.currentThread().setName(sourceTableName);
		//idscrement为一个批处理的处理的数量即id一次增加的数量
		idsIncrement = conf.getIdsIncrement();
		//读取上次汇总结束的地址做为本次汇总开始的地址
		idsBegin = readPointer(conf.getPointerTableName(), sourceTableName,
				conf.getTargetTableName());
		//读取数据来源表的数量
		idsEnd = maxId = gatherDao.getMaxId(sourceTableName);
		//idsEnd = maxId = 1000;
		//如果开始的ID小于结束的id就从idsBegin开始汇总
		while (idsBegin < idsEnd) {
			beginId = idsBegin;
			endId = beginId + idsIncrement - 1; // 取数据时两边都取等号
			if (endId <= maxId) {
				//如果上一批量结尾的id小于最大id说明汇总还没完成，就继续进行
				handleBatchData(beginId, endId, conf);
				idsBegin = idsBegin + idsIncrement;
			} else {
				endId = maxId; // endId应小于maxId
				handleBatchData(beginId, endId, conf);
				break;
			}
		}
		GatherProcess.gatherState.put(sourceTableName, false);
		
	}

	@Transactional(propagation=Propagation.REQUIRED)
	public void handleBatchData(int beginId, int endId, Configure conf) {
		logger.info("BeginId#" + sourceTableName + ":" + beginId);
		// 表示任务没有完成
		int maxId = gatherDao.getMaxId(sourceTableName);
		// 防止转移超过当前最大值的Id数据
		if (beginId >= 0 && endId > 0 && maxId >= endId) {
			// 更新执行开始时间
			logger.info("begin gathering...");
			
			// 插入Id段数据，忽略重复值
			try {
				logger.info("准备查询批量数据");
				String selectItems = getSelectItems(conf);
				//这里的目标表是relative_memos
				if (conf.getTargetTableName().equals("relative_memos")) {
					logger.info("对帖子进行汇总");
					
					List<RelativeMemo> dataGet = gatherDao.getPostGatherData(sourceTableName, selectItems, beginId, endId,
						                                                  conf.getAndWhere());
					logger.info("从数据源表中提取到数据");
					gatherPosts(dataGet);
					
				} else 
					//这是对项目进行汇总的操作
					if (conf.getTargetTableName().equals("gather_projects")) {
						List<GatherProject> dataGet = null;
						if(sourceTableName.equals("github"))
							dataGet = gatherDao.getPrjGatherDataForGit(sourceTableName, selectItems, beginId, endId,
										                                                 conf.getAndWhere());
						else
							dataGet = gatherDao.getPrjGatherData(sourceTableName, selectItems, beginId, endId,
										                                                 conf.getAndWhere());
						gatherProjects(dataGet);

					} else {
							List<JobRequirement> dataGet = gatherDao.getJobGatherData(
									sourceTableName, selectItems, beginId, endId,
									conf.getAndWhere());
							gatehrRequirement(dataGet);
						}
						
			} catch (Exception ex) {
						// 数据迁移过程可能发生异常情况
				        ex.printStackTrace();
						logger.error(ex);
						System.exit(0);
			}
		
					// 更新游标到本次 EndId+1;
					pointerDao.updatePointer(conf.getPointerTableName(),
							sourceTableName, conf.getTargetTableName(), endId + 1);// sourceIdBegin
																					// +
																					// idsIncrement
					logger.info("current--" + sourceTableName + ": " + endId);
		}
	}

	// 处理URL不存在的帖子 插入relative_memos表
	public void handleInsertGatherPosts(RelativeMemo model, Configure conf) {
		try {
			targetDao.insertRelativeMemo(conf.getTargetTableName(),
					conf.getTargetFields(), model);
		} catch (Exception e) {
			logger.error(e);
		}
	}

	// 处理URL相同的帖子更新 id表示更新的帖子固定id//更新url_md5相同的帖子
	public void handleUpdateGatherPosts(int id, RelativeMemo model_new) {
		targetDao.updateRelativeMemo(gatherPostsTableName, model_new, id);// 更新数据relative_memos表
	}
	public void handleUpdateGatherPostsByUrlMD5(int id, RelativeMemo model_new) {
		targetDao.updateRelativeMemoByUrlMD5(gatherPostsTableName, model_new, id);// 更新数据relative_memos表
	}

	// 处理URL不存在的项目 插入gather_projects表
	public void handleInsertGatherProjects(GatherProject model, Configure conf) {
		try {
			targetDao.insertOpenSourceProject(conf.getTargetTableName(),
					conf.getTargetFields(), model);
		} catch (Exception e) {
			logger.error(e);
		}

	}

	// 处理URL相同的项目更新 id表示更新的项目固定id
	public void handleUpdateGatherProjects(int id, GatherProject model_new) {
		targetDao.updateOpenSourceProject(conf.getTargetTableName(), model_new,
				id);// 更新数据gather_projects表
	}

	// 处理URL不存在的项目 插入job_requirements表
	public void handleInsertGatherJobs(JobRequirement model, Configure conf) {
		try {
			targetDao.insertJobRequirement(conf.getTargetTableName(),
					conf.getTargetFields(), model);
		} catch (Exception e) {
			logger.error(e);
		}

	}

	// 处理URL相同的项目更新 id表示更新的项目固定id
	public void handleUpdateGatherJobs(int id, JobRequirement model_new) {
		targetDao.updateJobRequirement(conf.getTargetTableName(), model_new, id);// 更新数据job_requirements表
	}
	
	// 将tag和项目的关系存入表item_tag_relation 并分离tag
	public void handleTags(RelativeMemo model,int postId)
	{
		// 将tag和项目的关系存入表item_tag_relation 并分离tag
			String tags = model.getTags();
			if(tags != null)
			{
			    List<String> tagList = DataHandler.tagsSegmentation(tags);
				for (String tag : tagList) {
					targetDao.insertTag(tagsTableName, tag);// ignore方式插入该项目的标签
					int tag_id = targetDao.selectTagIdByName(
							tagsTableName, tag);
					Taggings taggings = new Taggings();
					taggings.setTag_id(tag_id);
					taggings.setTaggable_id(postId);
					taggings.setTaggable_type("RelativeMemo");
					taggings.setContext("tags");
					taggings.setCreated_at(DataHandler.getNow());
					// 将Taggings对象存入数据库中
					try {
						targetDao.insertTaggings(memoTaggingsTableName,
								taggings);
					} catch (Exception e) {
						// 在插入记录之前 relative_memos表中的记录已经被删除掉了
						logger.error(e);
						System.exit(0);
					}
				}
			}
	}
	
	/**
	 * 最新修改的程序更好地考虑到了程序的有效性，主要分以下情况：
	 * 1.pk_control_posts中已存在（查询到pk_model），则先根据url_md5查询targetTable中是否存在改对象（设为model）。
	 * 如果model不为空，则判断pk_model.getId == model.getId是否成立，成立则直接更新，不成立则将pk_control中的
	 * 更新为model.getId.
	 * 2.pk_control_posts中不存在，则直接插入生成对象并再次查询到pk_model，先根据url_md5查询targetTable中是否存在改对象（设为model）。
	 * 如果model不为空，则判断pk_model.getId == model.getId是否成立，成立则直接更新，不成立则将pk_control中的
	 * 更新为model.getId.
	 * @param dataGet
	 */
	@Transactional(propagation=Propagation.REQUIRED)
	public void gatherPosts(List<RelativeMemo> dataGet){
		for (int i = 0; i < dataGet.size(); i++) {
			RelativeMemo model = dataGet.get(i);
			String urlMD5 = model.getUrl_md5();// 通过urlMD5判断是不是已经存在该帖子、是否更新
			
			int postId = 0;
			logger.info("开始汇总：" + model.getId());
			
			//查找pk_control_posts表以判定此帖子是否已经存在
			PKControlPosts pkControlModel = pkControlPostsDao.selectItemByUrlMD5(
						                      pkControlPostsTableName, urlMD5);
			//帖子已经存在则进行更新操作
			if(pkControlModel != null){
				
				try{
					logger.info("帖子已经在pk_control_posts存在");
					RelativeMemo memosModel =  targetDao.selectRelativeMemosItemByUrlMD5(conf.getTargetTableName(), model.getUrl_md5());
					if(memosModel!=null)
					{
						logger.info("帖子确实已经插入到了relative_memos表中");
						try
						{
							if(memosModel.getId() == pkControlModel.getId())
							{
								logger.error("memosModel.getId() == pkControlModel.getId()");
								model.setId(pkControlModel.getId());
						        handleUpdateGatherPosts(pkControlModel.getId(), model);
							}
							else
							{
							  logger.error("memosModel.getId() ！= pkControlModel.getId()");
							  gatherDao.updatePkControlTable(pkControlPostsTableName,
								                            memosModel.getId(), memosModel.getUrl_md5());
							  pkControlModel.setId(model.getId());
							  model.setId(memosModel.getId());
							  targetDao.deleteWhileExist(gatherPostsTableName,model);
							  handleUpdateGatherPostsByUrlMD5(pkControlModel.getId(), model);
					        }
							
						     logger.info("更新完成");
						}
						catch(Exception e)
						{
							logger.error(e);
							logger.error("更新失败：295");
								 System.exit(0);
						}
					}
					else
					{
						logger.info("帖子实际上并未插入到relative_memos表中");
						try{
							model.setId(pkControlModel.getId());
							targetDao.deleteWhileExist(gatherPostsTableName,model);
							handleInsertGatherPosts(model, conf);
							logger.info("更新转为插入");
						}
						catch (Exception e)
						{
							logger.error("更新转为插入时出错：298");
							System.exit(0);
							
						}
					}
						
				}catch(Exception e){
					logger.error("更新帖子时出错:" + e);
					System.exit(0);
				}
			}
			else
			{//帖子不存在就插入pk_control_posts表并为帖子生成唯一固定ID
				logger.info("准备将帖子插入pk_control_posts");
				try{
				pkControlPostsDao.insertOneItem(pkControlPostsTableName, urlMD5);
				
				}catch(Exception e)
				{
					logger.info("插入pkcontorl时失败");
					//System.exit(0);
					continue;
				}
				PKControlPosts controlItem = pkControlPostsDao.selectItemByUrlMD5(
						                     pkControlPostsTableName, urlMD5);
				
				try{
					RelativeMemo memosModel2 = targetDao.selectRelativeMemosItemByUrlMD5(conf.getTargetTableName(), model.getUrl_md5());
					
					if(memosModel2 == null)
					{
						 model.setId(controlItem.getId());
						 targetDao.deleteWhileExist(gatherPostsTableName,model);
					     handleInsertGatherPosts(model, conf);
					}
					else 
					{
					   if(controlItem.getId() == memosModel2.getId())
					   {
						   model.setId(controlItem.getId());
					       int sucess2 = targetDao.updateRelativeMemoByUrlMD5(gatherPostsTableName, model, model.getId());
					       if(sucess2 != 1)
					       {
					    	   logger.error("365");
					    	   System.exit(0);
					       }
					   }
					   else 
					   {
					      gatherDao.updatePkControlTable(pkControlPostsTableName, memosModel2.getId(), memosModel2.getUrl_md5());
					      model.setId(memosModel2.getId());
					      controlItem.setId(memosModel2.getId());
					      targetDao.deleteWhileExist(gatherPostsTableName,model);
					      handleUpdateGatherPosts(controlItem.getId(), model);
					      
					   }
					
					}
					logger.info("插入relative_memos成功");
				}catch(Exception e){
					logger.error("插入帖子时出错:" + e);
					int sucess = targetDao.updateRelativeMemoByUrlMD5(gatherPostsTableName, model, model.getId());
					if(sucess != 1)System.exit(0);
				}	
			}
			
			postId = model.getId();
			// 将tag和项目的关系存入表item_tag_relation 并分离tag
			handleTags(model,postId);
			
		}
	}
	@Transactional(propagation=Propagation.REQUIRED)
	public void gatherProjects(List<GatherProject> dataGet){
			for (int i = 0; i < dataGet.size(); i++) {
				GatherProject model = dataGet.get(i);
				String urlMD5 = model.getUrl_md5();// 通过urlMD5判断是不是已经存在该项目、是否更新
													 
				int prjId = 0;
				
				PKControlProjects pkControlProjects = pkControlProjectsDao.selectItemByUrlMD5(
								                           pkControlProjectsTableName, urlMD5);//查看有没有固定的id
				
				if(pkControlProjects != null){
					model.setId(pkControlProjects.getId());
					model.setUpdate_mark(1);
					
					try{
						if(targetDao.selectOpenSourceProjectsItem(conf.getTargetTableName(), model.getId())!=null)
							handleUpdateGatherProjects(pkControlProjects.getId(), model);
						else{
							model.setUpdate_mark(0);
							handleInsertGatherProjects(model, conf);
						}
					}catch(Exception e){
						logger.error("更新项目时出错:" + e);
					}
					
				}else{
					// 在pk_control_posts表中生成当前项目对应的id
					pkControlProjectsDao.insertOneItem(
							pkControlProjectsTableName, urlMD5);
					// 查看刚刚插入信息的id
					PKControlProjects controlItem = pkControlProjectsDao.selectItemByUrlMD5(
									                      pkControlProjectsTableName,urlMD5);
					// 用id构造model对应的固定不变的id
					model.setId(controlItem.getId());
					
					model.setUpdate_mark(0);
					try{
						handleInsertGatherProjects(model, conf);
					}catch(Exception e){
						logger.error("插入项目时出错:" + e);
					}
				}
				
				prjId = model.getId();
			}
	}
	
	@Transactional(propagation=Propagation.REQUIRED)
	public void gatehrRequirement(List<JobRequirement> dataGet){
		for (int i = 0; i < dataGet.size(); i++) {
			JobRequirement model = dataGet.get(i);
			String urlMD5 = model.getUrl_md5();// 通过urlMD5判断是不是已经存在该帖子
												// 是否更新
			int postId = 0;
			
			PKControlPosts pkControlPosts = pkControlPostsDao
			.selectItemByUrlMD5(
					pkControlPostsTableName, urlMD5);// 查看有没有固定的id
			
			if (pkControlPosts != null){
				model.setId(pkControlPosts.getId());
				
				try{
					handleUpdateGatherJobs(pkControlPosts.getId(), model);
				}catch(Exception e){
					logger.error("更新jobRequirement时出错:" + e);
				}
			}
			else {
				// 在pk_control_posts表中生成当前项目对应的id
				pkControlPostsDao.insertOneItem(
						pkControlPostsTableName, urlMD5);
				// 查看刚刚插入信息的id
				PKControlPosts controlItem = pkControlPostsDao
						.selectItemByUrlMD5(
								pkControlPostsTableName, urlMD5);
				// 用id构造model对应的固定不变的id
				model.setId(controlItem.getId());
				
				try{
					handleInsertGatherJobs(model, conf);
				}catch(Exception e){
					logger.error("插入jobRequirement时出错:"+ e);
				}
			}
			postId = model.getId(); 
		}
	}
	
	public String getSelectItems(Configure conf){
		String[] sourceFields = conf.getSourceFields().split(",");
		String[] targetFields = conf.getTargetFields().split(",");
		String selectItems = "";
		for (int i = 0; i < sourceFields.length; i++) {
			String str_source = sourceFields[i];
			String str_target = targetFields[i];
			selectItems += str_source + " as " + str_target + ",";
		}
		selectItems = selectItems
				.substring(0, selectItems.length() - 1) + " ";
		return selectItems;
	}
	
}
